package com.shujia.dwi

import org.apache.spark.sql.expressions.{UserDefinedFunction, Window}
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}
import com.shujia.common.utils.{Geography, SparkTool}
import org.apache.spark.internal.Logging

object DwiResRegnMergelocationMskDay extends SparkTool {


  /**
   * 构建区域位置数据融合表
   * 1、进行超速/乒乓清洗(基于速度进行过滤)
   * 2、数据脱敏
   */
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 加载ods层的数据
    val oiddDF: DataFrame = spark.table("ods.ods_oidd")

    oiddDF
      // 进行分区过滤，避免全表扫描
      .where($"day_id" === day_id)
      .withColumn("start_t", split($"start_time", ",")(1))
      .withColumn("end_t", split($"start_time", ",")(0))
      // 在时间轴上进行聚类：合并同一个网格内相邻的位置记录，解决同一个网格内相邻数据之间存在时间交叠的问题
      // 1、取上一条数据的网格id
      .withColumn("last_grid_id", lag($"grid_id", 1) over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 2、判断网格id是否相同，打上标记flag
      .withColumn("flag", when($"grid_id" === $"last_grid_id", 0).otherwise(1))
      // 3、对flag进行累加 得到grp
      .withColumn("grp", sum($"flag") over Window.partitionBy($"mdn").orderBy($"start_t"))
      // 4、基于grp合并同一个网格内连续的位置记录
      .groupBy($"mdn", $"grid_id", $"longi", $"lati", $"county_id", $"bsid", $"grp")
      .agg(min($"start_t") as "min_start_t", max($"end_t") as "max_end_t")

      // 取上一条数据的开始时间（为了避免不同网格的相邻数据当时间有交叠时出现负数的情况）
      .withColumn("last_start_t", lag($"min_start_t", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      // 计算时间差 单位：秒
      .withColumn("diff_time", unix_timestamp($"min_start_t", "yyyyMMddHHmmss") - unix_timestamp($"last_start_t", "yyyyMMddHHmmss"))

      // 取上一条数据的经纬度
      .withColumn("last_lg", lag($"longi", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      .withColumn("last_lat", lag($"lati", 1) over Window.partitionBy($"mdn").orderBy($"min_start_t"))
      // 计算两个经纬度之间的距离 单位：米
      .withColumn("distance", calculateLength($"longi", $"lati", $"last_lg", $"last_lat"))
      // 计算速度
      .withColumn("speed", when(round($"distance" / $"diff_time", 2).isNull, 0).otherwise(round($"distance" / $"diff_time", 2)))
      // 进行超速过滤
      /**
       * 速度定义：
       * 步行：1m/s
       * 骑行：5~10m/s
       * 自驾：15~40m/s
       * 火车：30~50m/s
       * 高铁：80~100m/s
       * 飞机：200~340m/s
       */
      .where($"speed" < 340)
      // 整理数据
      // 对手机号加上后缀（加盐）并计算md5
      .select(
        upper(md5(concat($"mdn", expr("'shujia'")))) as "mdn"
        , $"min_start_t" as "start_date"
        , $"max_end_t" as "end_date"
        , $"county_id"
        , $"longi"
        , $"lati"
        , $"bsid"
        , $"grid_id")
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dwi/dwi_res_regn_mergelocation_msk_d/day_id=$day_id")

    // 添加分区
    spark.sql(
      s"""
         |alter table dwi.dwi_res_regn_mergelocation_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)

  }


  /**
   * 任务提交的步骤
   * 1、打包并上传，需要将common模块生成的jar包一起上传
   * 2、进行授权
   * 通过root用户执行：hdfs dfs -chmod 777 /user/spark/applicationHistory
   * 通过ods用户执行：hdfs dfs -setfacl -R -m user:dwi:r-x /daas/motl/ods
   * 通过root用户执行：hdfs dfs -setfacl -m user:dwi:rwx /user
   * 3、提交任务，注意：需要引用common的jar包、还需要传入分区日期
   * spark-submit \
   * --master yarn-cluster \
   * --class com.shujia.dwi.DwiResRegnMergelocationMskDay  \
   * --conf spark.sql.shuffle.partitions=15 \
   * --jars common-1.0-SNAPSHOT.jar \
   * dwi-1.0-SNAPSHOT.jar \
   * 20221104
   */

}
